package org.ghost.akka.cluster.worker;

import akka.actor.AbstractActor;
import akka.cluster.Cluster;
import akka.cluster.ClusterEvent.CurrentClusterState;
import akka.cluster.ClusterEvent.MemberUp;
import akka.cluster.Member;
import akka.cluster.MemberStatus;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import org.ghost.akka.cluster.Constants;
import org.ghost.akka.common.command.StartCommand;
import org.ghost.akka.common.command.StopCommand;
import org.ghost.akka.common.em.ResponseEnum;
import org.ghost.akka.common.model.JobResult;

/**
 * 任务处理线程
 */
public class TaskWorker extends AbstractActor {
    LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    //集群
    Cluster cluster = Cluster.get(getContext().system());

	/*private TaskManageService tmService;
    private TaskManageService untmService;*/

    // subscribe to cluster changes, MemberUp
    @Override
    public void preStart() {
        cluster.subscribe(self(), MemberUp.class);
        /*tmService = ApplicationContextWrapper.getInstance().getSpringContext()
                .getBean("TaskManageService", TaskManageService.class);
		untmService = ApplicationContextWrapper.getInstance()
				.getSpringContext()
				.getBean("UnTaskManageService", TaskManageService.class);*/
    }

    // re-subscribe when restart
    @Override
    public void postStop() {
        cluster.unsubscribe(self());
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(StartCommand.class, command -> {
                    String taskId = command.getJob().getTaskId();
                    log.info("backed do [StartCommand] taskId=" + taskId);
                    log.info("start task...");
                    getSender().tell(new JobResult(ResponseEnum.JOB_SUCCESS, "开始成功"), self());
                })
                .match(StopCommand.class, command -> {
                    String taskId = command.getJob().getTaskId();
                    log.info("backed do [StopCommand] taskId=" + taskId);
                    log.info("stop task...");
                    getSender().tell(new JobResult(ResponseEnum.JOB_SUCCESS, "停止成功"), self());
                })
                .match(CurrentClusterState.class, state -> {
                    for (Member member : state.getMembers()) {
                        if (member.status().equals(MemberStatus.up())) {
                            register(member);
                        }
                    }
                })
                .match(MemberUp.class, mUp -> {
                    register(mUp.member());
                }).build();
    }

    void register(Member member) {
        if (member.hasRole("frontend")) {
            getContext().actorSelection(member.address() + "/user/frontend")
                    .tell(Constants.BACKEND_REGISTRATION, self());
        }
    }
}
